package com.gqr.springcloud.biz;

import com.gqr.springcloud.topics.*;
import lombok.extern.slf4j.Slf4j;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.messaging.Message;
import org.springframework.messaging.handler.annotation.Header;

import java.util.concurrent.atomic.AtomicInteger;

@Slf4j
@EnableBinding(
        value = {
                Sink.class,
                MyTopic.class,
                GroupTopic.class,
                DelayedTopic.class,
                ErrorTopic.class,
                RequeueTopic.class,
                DlqTopic.class,
                FallbackTopic.class
        }
)
public class StreamConsumer {
    private AtomicInteger count = new AtomicInteger(1);

    @StreamListener(Sink.INPUT)
    public void consume(Object payload) {
        log.info("message consume successfully, payload={}", payload);
    }

    /**
     * 自定义消息的消费者
     *
     * @param payload
     */
    @StreamListener(MyTopic.INPUT)
    public void consumeMyMessage(Object payload) {
        log.info("MyMessage consume successfully, payload={}", payload);
    }

    // 消息分组 & 消费分区示例
    @StreamListener(GroupTopic.INPUT)
    public void consumeGroupMessage(Object payload) {
        log.info("Group Message consume successfully, payload={}", payload);
    }

    // 延迟消息示例
    @StreamListener(DelayedTopic.INPUT)
    public void consumeDelayedMessage(MessageBean bean) {
        log.info("Delayed message consumed successfully, payload={}", bean.getPayload());
    }

    // 异常重试(单机版)
    @StreamListener(ErrorTopic.INPUT)
    public void consumeErrorMessage(MessageBean bean) {
        log.info("Are you ok?");
        if (count.incrementAndGet() % 3 == 0) {
            log.info("fine, thank you and you {}", bean.getPayload());
            count.set(0);
        } else {
            log.info("what is your problem");
            throw new RuntimeException("I'm not ok");
        }
    }


    // 异常重试(联机版-重新入列)
    @StreamListener(RequeueTopic.INPUT)
    public void consumeRequeueErrorMessage(MessageBean bean) {
        log.info("Are you ok?");
        try {
            Thread.sleep(3000);
        } catch (InterruptedException e) {
        }
        throw new RuntimeException("I'm not ok");
    }

    // 死信队列
    @StreamListener(DlqTopic.INPUT)
    public void consumeDlqMessage(MessageBean bean) {
        log.info("Dlq - Are you ok?");
        if (count.incrementAndGet() % 3 == 0) {
            log.info("Dlq - fine, thank you and you {}", bean.getPayload());
        } else {
            log.info("Dlq - what is your problem");
            throw new RuntimeException("I'm not ok");
        }
    }


    // fallback + 升级版本
    @StreamListener(FallbackTopic.INPUT)
    public void goodbyeBadGuy(MessageBean bean,
                              @Header("version") String version
    ) {
        log.info("fallback - Are you ok?");
        if ("1.0".equalsIgnoreCase(version)) {
            log.info("fallback - fine, thank you and you {}", bean.getPayload());
        } else if ("2.0".equalsIgnoreCase(version)) {
            log.info("unsupported version");
            throw new RuntimeException("I'm not ok");
        } else {
            log.info("fallback - version={}", version);
        }
    }

    // 这个就是fallback
    @ServiceActivator(inputChannel ="fallback-topic.fallback-group.errors" )
    public void fallback(Message<?> message){
        log.info("fallback entered");
    }
}
